package ds_industry_2025.industry.gy_05.T3

import org.apache.spark.sql.{SparkSession, functions}
import org.apache.spark.sql.functions._

import java.util.Properties

/*
    4、编写Scala代码，使用Spark根据dwd层的fact_change_record表展示每一个设备最近第二次的状态（倒数第二次），时间字段选用
    change_start_time，如果设备仅有一种状态，返回该状态（一个设备不会同时拥有两种状态），存入MySQL数据库shtd_industry的
    recent_state表中（表结构如下），然后在Linux的MySQL命令行中根据设备id降序排序，查询出前5条，将SQL语句复制粘贴至客户端桌
    面【Release\任务B提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的
    任务序号下；
 */
object t4 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t4")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sq.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    spark.table("dwd.fact_change_record")
      .createOrReplaceTempView("change_record")

    //  todo 先给所有的机器分区进行排序
    val r1 = spark.sql(
      """
        |select distinct
        |changemachineid as machine_id,
        |row_number() over(partition by changemachineid order by changestarttime desc ) as row,
        |changestarttime as start_time,
        |changeendtime as end_time,
        |changerecordstate as state
        |from change_record as c
        |""".stripMargin)


    //  todo 由于在分组聚合之后会丢弃其他字段，所以我们需要将分组统计每个机器数据数量这一段代码单独拿出来
   val countDF= r1.groupBy(col("machine_id"))
      .agg(functions.count("*").as("total_number"))

    //  todo 将状态只有一次的机器的数据拿到
    val t1 = r1.join(countDF,"machine_id")
      .filter(col("total_number") === 1 && col("row") === 1)
      .select("machine_id", "state", "start_time", "end_time")

    //  todo 将状态存在多次的数据拿到
    val t2 = r1.join(countDF,"machine_id")
      .filter(col("total_number") > 1 && col("row") === 2)
      .select("machine_id", "state", "start_time", "end_time")

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    //  todo 将结果合并    下面转换数据类型是因为end_time的默认值不符合mysql的规则
    t1.unionAll(t2)
      .withColumn("end_time",col("end_time").cast("String"))
      .write.mode("overwrite")
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_industry?useSSL=false","recent_state",conn)



    spark.close()
  }

}
